package gu.simplemq.mqtt;

import static gu.simplemq.mqtt.MqttPoolLazys.POOLS;
import static gu.simplemq.mqtt.MqttPoolLazys.setDefaultInstance;

import java.net.URI;
import java.util.Properties;

import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;

import com.google.common.base.Strings;

import gu.simplemq.pool.BaseMQPool;
import gu.simplemq.utils.IntrospectionSupport;
/**
 * 延迟初始化的 {@link MqttClient}资源池(线程安全),使用方法:<br>
 * 通过 {@link #getDefaultInstance()} 和getInstance(...)系列静态方法获取{@link MqttPoolLazy}实例<br>
 * 通过{@link #apply()} 和 {@link #free()}方法实现{@link MqttClient}对象的申请和释放
 * @author guyadong
 *
 */
public class MqttPoolLazy extends BaseMQPool<MqttClient> implements MqttConstants{
	private final Properties parameters;
	private String serverURI;
	private String clientId;
	private int qos = DEFAULT_QOS;
	private final MqttConnectOptions connOpts = new MqttConnectOptions();
	private volatile MqttClient mqttClient;
	private final URI location;
	protected MqttPoolLazy (Properties props) {
		this.parameters=PropertiesHelper.MHELPER.initParameters(props);
		location = PropertiesHelper.MHELPER.getLocationlURI(props);
		POOLS.put(location,this);
		IntrospectionSupport.setProperties(this, parameters, false);
		IntrospectionSupport.setProperties(connOpts, parameters, CONNOPTS_PREFIX);
		// 设置自动重连接
		connOpts.setAutomaticReconnect(true);
	}
	
	public MqttPoolLazy setServerURI(String serverURI) {
		this.serverURI = serverURI;
		return this;
	}

	public MqttPoolLazy setClientId(String clientId) {
		this.clientId = clientId;
		return this;
	}

	public int getQos() {
		return qos;
	}

	public void setQos(int qos) {
		this.qos = qos;
	}

	public Properties getParameters() {
		return new Properties(parameters);
	}

	/**
	 * 将当前实例指定为默认实例
	 * @return
	 * @see #setDefaultInstance(MqttPoolLazy)
	 */
	public MqttPoolLazy asDefaultInstance(){
		setDefaultInstance(this);
		return this;
	}
	/**
	 * @see gu.simplemq.pool.BaseMQPool#borrow()
	 * @throws MqttPoolException 
	 */
	@Override
	public MqttClient borrow()throws MQPoolException{
		// double-checked locking
		if(mqttClient == null){
			synchronized (this) {
				if(mqttClient == null){
			        try {
			    		mqttClient = new MqttClient(
			    				serverURI, 
			    				Strings.isNullOrEmpty(clientId) ? MqttClient.generateClientId() : clientId, 
			    				null);
			    		mqttClient.connect(connOpts);
			    		logger.info("mqtt client initialized(MQTT客户端初始化)  {} ",getCanonicalURI());
					} catch (MqttException e) {
						throw new MQPoolException(e);
					}
				}
			}
		}
    	return mqttClient;
    }
    
	@Override
    public void release(MqttClient r) {
		// DO NOTHING
    }
	@Override
	public URI getCanonicalURI(){
		return PropertiesHelper.MHELPER.getLocationlURI(parameters);
	}
	
	@Override
	public void close(){
		// double check
		if(mqttClient != null){
			synchronized (this) {
				if(mqttClient != null){
					try {
						// 先要断开连接才能关闭client
						if(mqttClient.isConnected()){
							mqttClient.disconnect();
						}
						mqttClient.close();
						POOLS.remove(location);
						logger.info("discard mqtt client: {}",this);
						mqttClient = null;
						closed = true;
					} catch (MqttException e) {
						e.printStackTrace();
					}
				}				
			}
		}
	}
}
